package study.datastructure.queue.impl;/**
 * @program: datastructure
 * @author: lcy
 * @create: 2024-11-02 15:09
 * <p>
 * 2024/11/2,
 */

/**
 2024/11/2,
 */

import study.datastructure.queue.BlockingQueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单锁实现
 */
public class TestBlockingQueue1<E> implements BlockingQueue<E> {

    @SuppressWarnings("unchecked")
    public final  E[] queue;

    //实际大小
    private int size;

    //尾部
    int tail=0;

    //队列头
    int head=0;
    public TestBlockingQueue1(int capacity) {
        queue= (E[]) new Object[capacity];
    }

    private boolean isFull() {return size == queue.length;}

    private boolean isEmpty() { return size == 0; }

    ReentrantLock lock = new ReentrantLock();
    Condition tailsWait = lock.newCondition(); //阻塞线程的集合
    Condition headsWait = lock.newCondition(); //阻塞线程的集合


    @Override
    public void offer(E e) throws InterruptedException {
        lock.lock();
        try {
            while (isFull()) {
                tailsWait.await();// 让当前线程阻塞 直到被唤醒(直到队列不为空)
            }

            //插入
            queue[tail] = e;
            tail++;
            if (tail == queue.length) {
                tail = 0;
            }

            size++;

            //唤醒 poll取值线程
            headsWait.signal();

        } finally {
            lock.unlock();
        }
    }

    /**
     * 插入的时间范围内 没有插入成功 就返回FALSE 不再等待
     * @param e 插入值
     * @param timeout 超时时间
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        lock.lockInterruptibly(); //相比于lock 能及时响应打断调度
        long nanos= TimeUnit.SECONDS.toNanos(timeout);
        try {
            while (isFull()) {
                if (nanos <= 0) {
                    return false;
                }
               nanos= tailsWait.awaitNanos(nanos);// 让当前线程阻塞 直到被唤醒(直到队列不为空)
            }

            //插入
            queue[tail] = e;
            tail++;
            if (tail == queue.length) {
                tail = 0;
            }
            size++;

            //唤醒 需要取值线程
            headsWait.signal();

        } finally {
            lock.unlock();
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        lock.lockInterruptibly(); //相比于lock 能及时响应打断调度
        try {
            while (isEmpty()) {
                headsWait.await();// 让当前线程阻塞 直到被唤醒(直到队列不为空)
            }

            //取值
            E e = (E) queue[head];
            if (++head == queue.length) {
                head = 0;
            }
            size--;

            //唤醒 需要插入值线程
            tailsWait.signal();

        } finally {
            lock.unlock();
        }

        return null;
    }



    @Override
    public String toString() {
        return "TestBlockingQueue1{" +
                "queue=" + Arrays.toString(queue)+
                '}';
    }

    public static void main(String[] args) {
        // 创建一个容量为5的阻塞队列实例
        TestBlockingQueue1<Integer> blockingQueue = new TestBlockingQueue1<>(2);

        // 创建生产者线程
        Thread producerThread = new Thread(() -> {
            for (int i = 1; i <= 10; i++) {
                try {
                    blockingQueue.offer(i,2);
                    System.out.println("offer" + blockingQueue);

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    e.printStackTrace();
                }
            }
        },"offer");

        // 创建消费者线程
        Thread consumerThread = new Thread(() -> {
            for (int i = 1; i <= 2; i++) {
                try {
                     blockingQueue.poll();
                    System.out.println("poll" + blockingQueue);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    e.printStackTrace();
                }
            }
        },"poll");

        // 启动生产者和消费者线程
        producerThread.start();
        consumerThread.start();

        try {
            // 等待生产者和消费者线程完成任务
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }

        System.out.println("测试完成");
    }

}
